package com.zc.study

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala._
import collection.JavaConverters._
import org.apache.flink.configuration.Configuration
import org.apache.flink.core.fs.FileSystem.WriteMode

object BroadcastVariablesApp {
  def main(args: Array[String]): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment
    val dataSet = env.fromElements((1,"1"),(2,"2"),(3,"3"),(4,"4"))
    val broadcastSet = env.fromCollection(1 to 3)

    val handData = dataSet.map(new RichMapFunction[(Int,String),(Int,String)] {
      override def map(value: (Int,String)): (Int,String) = value

      override def open(parameters: Configuration): Unit = {
        val broadcastList = getRuntimeContext.getBroadcastVariable[Int]("broadcastSet")
        for(item <- broadcastList.asScala){
          println(item)
        }
      }
    }).withBroadcastSet(broadcastSet,"broadcastSet")

    val sinkFilePath = "Z:\\gitee\\study_imooc\\test-data\\flink\\sink-csv\\broadcase.csv"
    handData.writeAsCsv(sinkFilePath,writeMode = WriteMode.OVERWRITE)

    env.execute()

  }
}
